

package cn.spark.study.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds
import org.apache.spark.rdd.RDD.rddToOrderedRDDFunctions
import org.apache.spark.streaming.dstream.DStream.toPairDStreamFunctions

object WindHotWordScala {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("")
      .setMaster("local[2]")
      
      val ssc = new StreamingContext(conf,Seconds(2))
    
      val parihostWords = ssc.socketTextStream("spark1", 9999).map { hotwords => (hotwords.split(" ")(0),1) }
      
      val windowWordCount = parihostWords.reduceByKeyAndWindow((v1:Int,v2:Int) => {
        v1 + v2
      }, Seconds(60), Seconds(10))
      
      val SortedWordCount = windowWordCount.map(pair => (pair._2,pair._1)).transform(pairCountWord => {
        val sortedWordCount = pairCountWord.sortByKey(false).map(pairCountWord => (pairCountWord._2,pairCountWord._1))
        val top3HotWord = sortedWordCount.take(3)
        for(hw <- top3HotWord){
          print(hw._1 + ":" + hw._2)
        }
        
        sortedWordCount
      });
      
      SortedWordCount.print()
      ssc.start()
      ssc.awaitTermination()
  }
}